package org.rabbitmq.mq02.mqServer.datacore;

import org.rabbitmq.mq02.common.MqException;
import org.rabbitmq.mq02.mqServer.core.Binding;
import org.rabbitmq.mq02.mqServer.core.Exchange;
import org.rabbitmq.mq02.mqServer.core.MSGQueue;
import org.rabbitmq.mq02.mqServer.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 将数据存储在内存中,创建不同的数据集合来管理
 * 要管理的数据有:
 * 交换机
 * 队列
 * 绑定关系
 * 消息
 * 队列中的消息集合
 * 待确认消息队列中的消息集合
 */
public class MemoryDataCenter {
    //key:exchangeName
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    //key:queueName
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
//   key1:exchangeName key2:queueName
    private ConcurrentHashMap<String,ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    //key: messageId
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    //  key:queueName  List:message
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    //存储在手动确认模式下,管理待确认的消息和队列,在未收到确认消息时,要先将数据存储到这个数据集合中,
    //  key1:queueName   key2:messageId
    private ConcurrentHashMap<String,ConcurrentHashMap<String,Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    //交换机:
    //插入
    public void insertExchange(Exchange exchange){
        exchangeMap.put(exchange.getName(),exchange);
        System.out.println("[MemoryDataCenter] 新增交换机成功 exchangeName:"+exchange.getName());
    }
    //删除
    public void deleteExchange(String exchangeName){
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 删除交换机成功 exchangeName:"+exchangeName);
    }
    //查找
    public Exchange getExchange(String exchangeName){
        Exchange exchange = exchangeMap.get(exchangeName);
        return exchange;
    }

    //队列
    //插入
    public void insertQueue(MSGQueue queue){
        queueMap.put(queue.getName(),queue);
        System.out.println("[MemoryDataCenter] 新增队列成功! queueName: "+queue.getName());
    }
    //删除
    public void deleteQueue(String queueName){
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 队列删除成功! queueName: "+queueName);
    }
    //查找
    public MSGQueue getQueue(String queueName){
        return queueMap.get(queueName);
    }

    //绑定关系:
    //新增
    public void insertBinding(Binding binding) throws MqException {
//        //绑定关系不存在时,创建一个,存在时,进行覆盖
//        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getQueueName());
//        if(bindingMap==null){
//            bindingMap = new ConcurrentHashMap<>();
//        }
//        bindingMap.put(binding.getQueueName(),binding);
//        bindingsMap.put(binding.getExchangeName(),bindingMap);
        //这个方法是ConcurrentMap方法用来判断对应的哈希表是否存在,不存在就执行第二个参数,存在就直接赋值,和上面的逻辑是一样的
        //且该方法是原子的,不存在线程安全问题
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                f -> new ConcurrentHashMap<>());
        //此处可能会存在线程安全问题,以绑定关系为基准进行上锁
        synchronized (binding){
            Binding binding1 = bindingMap.get(binding.getQueueName());
            //当绑定关系已经存在时,抛出异常,只有新的绑定插入时,才会成功
            if(binding1!=null){
                throw new MqException("[MemoryDataCenter] 绑定已存在 exchangeName: "+binding.getExchangeName()
                        +" ,queueName: "+binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(),binding);
        }
        System.out.println("[MemoryDataCenter] 新的绑定创建成功! +exchangeName:"+binding.getExchangeName()
                +" ,queueName: "+binding.getQueueName()+" ,bindingKey: "+binding.getBindingKey());
    }
    //删除
    public void deleteBinding(Binding binding) throws MqException {
    //先判断交换机是否存在绑定,不存在时,无法删除.抛出异常
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if(bindingMap==null){
            throw new MqException("[MemoryDataCenter] 无绑定关系,删除失败 exchangeName: "+binding.getExchangeName()
                    +" ,queueName: "+binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 绑定删除成功 exchangeName: "+binding.getExchangeName()
                +" ,queueName: "+binding.getQueueName()+" ,bindingKey:"+binding.getBindingKey());
    }
    //查找
    public Binding getBinding(String exchangeName,String queueName) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if(bindingMap==null){
            return null;
        }
        Binding binding = bindingMap.get(queueName);
        return binding;
    }
    //查找交换机所有的绑定集合:
    public ConcurrentHashMap<String,Binding> getBindings(String exchangeName){
        return bindingsMap.get(exchangeName);
    }

    //消息
    //插入
    public void insertMessage(Message message){
        messageMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 新增消息成功 messageID:"+message.getMessageId());
    }
    //删除
    public void deleteMessage(String messageId){
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息删除陈功 messageId: "+messageId);
    }
    //查找
    public Message getMessage(String messageId){
        return messageMap.get(messageId);
    }

    //队列消息集合
    //发送消息到指定队列
    public void sendMessage(MSGQueue queue,Message message){
        //1.先查找队列对应的集合是否存在,不存在时创建消息集合
//        LinkedList<Message> messages = queueMessageMap.get(queue.getName());
//        if(messages==null){
//            messages = new LinkedList<>();
//        }
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), f -> new LinkedList<>());
        //这里当多个线程同时执行插入操作时,可能会覆盖消息,要以集合为维度进行上锁
        synchronized(messages){
            messages.add(message);
        }
        //将消息也存入到消息集合中
        messageMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 发送消息到队列成功 queueName:"+queue.getName()+
                " ,messageId:"+message.getMessageId());
    }
    //从队列中取消息
    public Message pollMessage(String queueName){
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages==null || messages.isEmpty()){
            return null;
        }
        Message message = messages.remove(0);
        System.out.println("[MemoryDataCenter] 从队列中取消息成功 queueName:"+queueName+
                " ,messageId:"+message.getMessageId());
        return message;
    }
    //获取队列中的消息个数
    public int getMessageCountFromQueue(String queueName){
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages==null) return 0;
        //此处获取集合中元素个数可能存在线程安全问题,对集合进行上锁
        synchronized(messages){
            return messages.size();
        }
    }

    //待确认消息集合
    //发送消息到待确认消息集合
    public void sendWaitMessage(String queueName,Message message){
        ConcurrentHashMap<String, Message> waitMessagesMap = queueMessageWaitAckMap.computeIfAbsent(queueName, f -> new ConcurrentHashMap<>());
        //此处向待确认消息集合中插入数据时,也可能存在线程安全问题,以集合为维度加锁
        synchronized (waitMessagesMap){
            waitMessagesMap.put(message.getMessageId(),message);
        }
        System.out.println("[MemoryDataCenter] 发送待确认消息到队列成功 queueName:"+queueName+
                " ,messageId:"+message.getMessageId());
    }
    //从队列中取待确认消息
    public Message pollWaitMessage(String queueName,String messageId){
        ConcurrentHashMap<String, Message> waitMessagesMap = queueMessageWaitAckMap.get(queueName);
        if(waitMessagesMap==null){
            return null;
        }
        Message message = waitMessagesMap.get(messageId);
        if(message==null){
            return null;
        }
        System.out.println("[MemoryDataCenter] 从队列中取代确认消息成功 messageId:"+messageId+
                " ,queueName:"+queueName);
        return message;
    }
    //从队列中删除待确认消息
    public void deleteWaitMessage(String queueName,String messageId){
        ConcurrentHashMap<String, Message> waitMessagesMap = queueMessageWaitAckMap.get(queueName);
        if(waitMessagesMap==null){
            System.out.println("[MemoryDataCenter] 待确认消息队列不存在,消息删除失败 messageId:"+messageId+
                    " ,queueName:"+queueName);
        }
        waitMessagesMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 待确认消息删除成功 messageId:"+messageId+
                " ,queueName:"+queueName);
    }

    //回复所有硬盘中的数据
    //当服务器重启后,内存中的数据都不存在了,要从磁盘中获取数据
    public void recovery(DiskDataManager diskDataManager) throws IOException, MqException, ClassNotFoundException {
        //先将内存中的集合都清空,防止存在残留数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        queueMessageWaitAckMap.clear();
        //恢复交换机数据
        List<Exchange> exchanges = diskDataManager.selectAllExchanges();
        for(Exchange e:exchanges){
            String exchangeName = e.getName();
            exchangeMap.put(exchangeName,e);
        }
        //恢复队列数据
        List<MSGQueue> queues = diskDataManager.selectAllQueues();
        for(MSGQueue q:queues){
            String queueName = q.getName();
            queueMap.put(queueName,q);
        }
        //恢复绑定关系
        List<Binding> bindings = diskDataManager.selectAllBindings();
        for(Binding b:bindings){
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(b.getExchangeName(), f -> new ConcurrentHashMap<>());
            bindingMap.put(b.getQueueName(),b);
        }

        //恢复消息
        for(MSGQueue q:queueMap.values()){
            List<Message> messages = diskDataManager.loadAllMessageFromQueue(q.getName());
            for(Message m:messages){
                messageMap.put(m.getMessageId(),m);
            }
        }
        //对于未确认消息,当服务器重启后,服务器中所有的消息都要重新发送,未被确认的消息就都成了未被取走的消息了,
        //对于未确认的消息, 就不需要回复这些数据了
    }
}
